package com.atguigu.app

import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventLog}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.util.control.Breaks._

object AlertApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AlerApp")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    //3.消费kafka中用户行为数据的事件日志
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT, ssc)

    //4.将读取到的数据转为样例类
    val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
    val eventLogDStream = kafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])

        //补全字段
        val times: String = sdf.format(new Date(eventLog.ts))
        eventLog.logDate = times.split(" ")(0)
        eventLog.logHour = times.split(" ")(1)
        (eventLog.mid, eventLog)
      })
    })

    //5.开启一个5min的窗口
    val midTowindowDStream = eventLogDStream.window(Minutes(5))

    //6.对相同mid的数据聚合到一块
    val midToIterDStream: DStream[(String, Iterable[EventLog])] = midTowindowDStream.groupByKey()

    //7.根据条件筛选数据
    /**
      * 三次及以上用不同账号登录并领取优惠劵，（先判断用户是否领优惠券，然后对领优惠券的用户做去重，使用set集合做去重，然后判断set集合的大小是否大于等于三）
      * 并且过程中没有浏览商品。（判断用户是否浏览商品，如果浏览商品的话，则不符合以上要求，可以直接跳过）
      */
    val boolToCouponAlertInfoDStream: DStream[(Boolean, CouponAlertInfo)] = midToIterDStream.mapPartitions(partition => {
      partition.map { case (mid, iter) =>
        //创建用来存放领优惠券的用户id
        val uids: util.HashSet[String] = new util.HashSet[String]()
        //创建用来存放领优惠券所涉及的商品id
        val itemIds: util.HashSet[String] = new util.HashSet[String]()
        //创建用来存放用户涉及事件的集合
        val events: util.ArrayList[String] = new util.ArrayList[String]()

        //定义标志位用来判断这相同mid五分钟有没有浏览商品行为
        var bool: Boolean = true

        //a.先遍历迭代器中每条数据
        breakable {
          iter.foreach(log => {
            events.add(log.evid)
            //b.判断迭代器中（相同mid5分钟的数据）是否有用户浏览商品
            if ("clickItem".equals(log.evid)) {
              //有浏览商品行为,直接跳过（）
              bool = false
              break()
            } else if ("coupon".equals(log.evid)) {
              //没有浏览商品但是有领优惠券
              uids.add(log.uid)
              itemIds.add(log.itemid)
            }
          })
        }
        //生产疑似预警日志
        (uids.size() >= 3 && bool, CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis()))
      }
    })

    //8.生产预警日志
    val couponAlertInfoDStream: DStream[CouponAlertInfo] = boolToCouponAlertInfoDStream.filter(_._1).map(_._2)

    couponAlertInfoDStream.print()

    //9.将预警日志写入ES
    //并且同一设备，每分钟只记录一次预警（es的幂等性来去重 -》文档id怎么设置？mid+精确到分钟的时间）
    couponAlertInfoDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, CouponAlertInfo)] = partition.toList.map(log => {
          (log.mid + log.ts / 1000 / 60, log)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_INDEX_ALERT+"220212",list)
      })
    })

    //10.启动并阻塞
    ssc.start()
    ssc.awaitTermination()
  }

}
